#include "thread_pool.h"

// 创建线程个数  任务队列的大小
tpool_t *thread_pool_create(int cnt, int queueSize)
{
    int err, ret;

    tpool_t *pool = (tpool_t *)malloc(sizeof(tpool_t));
    if (!pool)
        goto err;

    // 工作线程ID集合
    pool->tp_work_thread_ids = (pthread_t *)malloc(sizeof(pthread_t) * cnt);
    if (!pool->tp_work_thread_ids)
        goto err;
    memset(pool->tp_work_thread_ids, 0, sizeof(pthread_t) * cnt);

    pool->tp_number_of_threads = cnt;

    ret = pthread_mutex_init(&pool->tp_mutex_pool, NULL);
    if (ret != 0)
        goto err;

    ret = pthread_cond_init(&pool->tp_cond_empty, NULL);
    if (ret != 0)
        goto err;

    ret = pthread_cond_init(&pool->tp_cond_full, NULL);
    if (ret != 0)
        goto err;

    pool->tp_shutdown = false;

    pool->tp_qcapacity = queueSize;
    pool->tp_qsize = 0;

    pool->tp_front = 0;
    pool->tp_rear = 0;

    // 任务结构体
    pool->tp_task = (tpool_task_t *)malloc(sizeof(tpool_task_t) * queueSize);
    if (!pool->tp_task)
        goto err;

    // 创建线程
    for (int i = 0; i < pool->tp_number_of_threads; i++)
    {
        // printf("%s,%s,%d\n", __FILE__, __FUNCTION__, __LINE__);
        ret = pthread_create(&pool->tp_work_thread_ids[i], NULL, tp_worker, pool);
        if (ret != 0)
            goto err;

        // 线程分离
        pthread_detach(pool->tp_work_thread_ids[i]);
    }

    return pool;
err:
    if (pool && pool->tp_task)
        free(pool->tp_task);
    if (pool && pool->tp_work_thread_ids)
        free(pool->tp_work_thread_ids);
    if (pool)
        free(pool);

    return NULL;
}

// 线程池任务队列中添加
// pool:线程池 task:实际服务端处理函数 arg:处理函数参数，服务端接收到的数据包
void thread_pool_add_tasks(tpool_t *pool, void (*task)(void *arg), void *arg)
{
    pthread_mutex_lock(&pool->tp_mutex_pool);

    //   判断队列是否已经满 线程未被销毁
    while (pool->tp_qcapacity == pool->tp_qsize && !pool->tp_shutdown)
    {
        // 1.解锁
        // 2.让线程阻塞在条件变量上
        // 3.在条件变量收到唤醒信号，线程唤醒，重新竞争锁，获得锁
        pthread_cond_wait(&pool->tp_cond_full, &pool->tp_mutex_pool);
    }

    //   线程池是否销毁
    if (pool->tp_shutdown)
    {
        pthread_mutex_unlock(&pool->tp_mutex_pool);
        return;
    }

    //   任务函数添加到任务队列
    pool->tp_task[pool->tp_rear].function = task;
    pool->tp_task[pool->tp_rear].arg = arg;
    pool->tp_rear = (pool->tp_rear + 1) % pool->tp_qcapacity;
    pool->tp_qsize++;

    //   唤醒一个工作线程执行任务
    pthread_cond_signal(&pool->tp_cond_empty);

    pthread_mutex_unlock(&pool->tp_mutex_pool);
}

// 线程执行
void *tp_worker(void *arg)
{
    tpool_t *pool = (tpool_t *)arg;
    tpool_task_t task;

    for (;;)
    {
        pthread_mutex_lock(&pool->tp_mutex_pool);

        // 判断队列是否已经空 线程池未被销毁
        while (pool->tp_qsize == 0 && !pool->tp_shutdown)
        {
            // 阻塞在条件变量
            pthread_cond_wait(&pool->tp_cond_empty, &pool->tp_mutex_pool);
        }

        // 线程池是否销毁
        if (pool->tp_shutdown)
        {
            pthread_mutex_unlock(&pool->tp_mutex_pool);
            pthread_exit(NULL);
        }

        // 从任务队列中获取一个任务
        task.function = pool->tp_task[pool->tp_front].function;
        task.arg = pool->tp_task[pool->tp_front].arg;
        pool->tp_front = (pool->tp_front + 1) % pool->tp_qcapacity;
        pool->tp_qsize--;

        pthread_cond_signal(&pool->tp_cond_full);

        pthread_mutex_unlock(&pool->tp_mutex_pool);

        printf("%s,%s,%d thread:%ld\n", __FILE__, __FUNCTION__, __LINE__, pthread_self());
        // 执行任务
        task.function(task.arg);

        free(task.arg);
    }
}

int thread_pool_destroy(tpool_t *pool)
{
    if (!pool)
    {
        return -1;
    }

    pool->tp_shutdown = true;

    // 唤醒阻塞的工作线程 设置销毁为true时，线程退出
    for (int i = 0; i < pool->tp_number_of_threads; i++)
    {
        pthread_cond_signal(&pool->tp_cond_empty);
    }

    // 释放任务队列
    if (pool->tp_task)
        free(pool->tp_task);

    // 释放保存线程ID空间
    if (pool->tp_work_thread_ids)
        free(pool->tp_work_thread_ids);

    // 释放线程池数据结构
    free(pool);
    pool = NULL;

    return 0;
}